

namespace Stee.CAP8.MessageBus.Kafka.Publisher
{
    using System;
    using System.Collections.Generic;
    using System.Configuration;
    using System.Linq;
    using System.Text;
    using System.Threading;
    using System.Threading.Tasks;
    using Confluent.Kafka;
    using NLog;
    using Stee.CAP8.MessageBus.Interfaces.Delegates;
    using Stee.CAP8.MessageBus.Interfaces.Enums;
    using Stee.CAP8.MessageBus.Interfaces.Interfaces;
    using Stee.CAP8.MessageBus.Interfaces.Models;
    using Stee.CAP8.MessageBus.Interfaces.Utility;
    using Stee.CAP8.MessageBus.Kafka.Configurations;
    using Stee.CAP8.MessageBus.Kafka.Constants;
    using Stee.CAP8.MessageBus.Kafka.Utility;

    /// <summary>
    /// Implements the publisher/producer interface for publishing messages.
    /// 
    /// https://github.com/confluentinc/confluent-kafka-dotnet/wiki/Producer
    /// Avoid producing synchronously if at all possible
    ///     You should be aware that producing synchronously will reduce maximum throughput dramatically.
    ///     For example, you might achieve a rate of about 500,000 msg/s if you don't wait on the result of each produce call
    ///     before sending the next. If you do, you'll get at best ~300 msg/s. This means you should typically not await your
    ///     ProduceAsync calls, the exception being if you are in a highly concurrent scenario (such as a web request handler).
    ///     In this case, although an individual call my be slow, it will not impact other calls happening simultaneously.
    /// 
    /// ProduceAsync & Produce
    ///     The two methods are equivalent, but tailored to different usage patterns.
    ///     The Produce method is more efficient, and you should care about that if your throughput is high (>~ 20k msgs/s),
    ///     else the difference will be negligible compared to whatever else you application is doing. Another difference is that
    ///     tasks returned by ProduceAsync complete on thread pool threads, so ordering is not guaranteed. By comparison,
    ///     delivery report callbacks passed to Produce are all called on the same thread, so the ordering will exactly match
    ///     the order they are known to be completed by the client.
    /// </summary>
    public sealed class PublisherImpl : IPublisher
    {
        #region Constant & Static Variables

        private readonly static Logger Logger = LogManager.GetCurrentClassLogger();

        /// <summary>
        /// Time span for waiting producer to flush messages
        /// </summary>
        private readonly static TimeSpan TimeoutProducerFlush = TimeSpan.FromMilliseconds(ProducerFlushTimeout);

        /// <summary>
        /// Timeout delay
        /// </summary>
        public const int ProducerFlushTimeout = 5000;

        #endregion Constant & Static Variables


        #region Fields

        /// <summary>
        /// Kafka producer configuration
        /// </summary>
        private ProducerConfig producerConfig;

        /// <summary>
        /// Kafka producer for publishing message
        /// </summary>
        private IProducer<string, byte[]> producer;

        /// <summary>
        /// Response handler for publishing messages
        /// </summary>
        private HandlePublishResponse publishResponseHandler;

        /// <summary>
        /// Publisher error handler
        /// </summary>
        private HandleBrokerError publishErrorHandler;

        /// <summary>
        /// Mutex lock
        /// </summary>
        private readonly object mutex = new object();

        #endregion Fields


        #region Properties

        /// <summary>
        /// Gets the flag indicating that publisher is initialized and ready for publishing
        /// </summary>
        public bool IsInitialized { get; private set; } = false;

        /// <summary>
        /// Gets the flag indicating if publisher is disposed
        /// </summary>
        public bool IsDisposed { get; private set; } = false;

        #endregion Properties


        #region Constructors & Initializers

        /// <summary>
        /// Initializes a new instance of the PublisherImpl class
        /// </summary>
        public PublisherImpl() { }

        #endregion Constructors & Initializers


        #region Interface Implementation

        /// <summary>
        /// Initializes the publisher
        /// </summary>
        /// <param name="brokerConfigurationJSON">JSON configuration path</param>
        /// <param name="publishResponseHandler">Publish response handler, response data type</param>
        /// <param name="brokerErrorHandler">Error handler</param>
        /// <returns>Return TRUE if initialized else FALSE</returns>
        public bool Initialize(string brokerConfigurationJSON, HandlePublishResponse publishResponseHandler, HandleBrokerError brokerErrorHandler)
        {
            try
            {
                if (this.IsInitialized)
                {
                    throw new InvalidOperationException($"Publisher is already initialized. IsInit: {this.IsInitialized}");
                }
                else if (publishResponseHandler == null)
                {
                    throw new ArgumentNullException($"Publish message reponse handler is null: {nameof(publishResponseHandler)}");
                }
                else if (brokerErrorHandler == null)
                {
                    throw new ArgumentNullException($"Error handler for broker is null: {nameof(brokerErrorHandler)}");
                }

                this.publishResponseHandler = publishResponseHandler;
                this.publishErrorHandler = brokerErrorHandler;

                var configs = ConfigurationHelper.GetBrokerConfiguration(brokerConfigurationJSON, ConfigurationConstants.BrokerConfigPublishSection);

                if (configs?.Any() == true)
                {
                    var configuration = new PublisherConfiguration(configs);

                    this.producerConfig = new ProducerConfig
                    {
                        BootstrapServers = configuration.BootstrapServers,
                        ClientId = configuration.ClientID,
                        Acks = configuration.Acks,
                        EnableIdempotence = configuration.EnableIdempotence,
                        LingerMs = configuration.LingerMs,
                        MessageSendMaxRetries = configuration.MessageSendMaxRetries,
                        MessageTimeoutMs = configuration.MessageTimeoutMs,
                        QueueBufferingMaxKbytes = configuration.QueueBufferingMaxKbytes,
                        QueueBufferingMaxMessages = configuration.QueueBufferingMaxMessages,
                        TransactionTimeoutMs = configuration.TransactionTimeoutMs,
                    };

                    this.InitializeProducer();
                    this.IsInitialized = true;

                    return this.IsInitialized;
                }
                else
                {
                    throw new ConfigurationErrorsException($"No configuration found for producer in section: {ConfigurationConstants.BrokerConfigAdminSection}");
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Publish generic message
        /// </summary>
        /// <param name="topicName">Message topic</param>
        /// <param name="genericMsg">Generic message</param>
        /// <param name="cancelToken">Cancel token</param>
        /// <returns>Return TRUE if published successfully else FALSE</returns>
        public bool PublishGeneric(TopicName topicName, GenericMessage genericMsg, CancellationToken cancelToken)
        {
            try
            {
                this.CheckProducerState();
                ValidateHelper.ValidateTopicName(topicName);
                ValidateHelper.ValidateGeneric(genericMsg);

                cancelToken.ThrowIfCancellationRequested();
                return this.ProduceGeneric(topicName, genericMsg, cancelToken);
            }
            catch (OperationCanceledException ex)
            {
                Logger.Info(ex, ex.Message);
                return false;
            }
            catch (ProduceException<string, byte[]> ex)
            {
                Logger.Error(ex, $"{ex.Error.Reason}: {ex.Message}");
                return false;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Publish generic message asynchronously
        /// </summary>
        /// <param name="topicName">Message topic</param>
        /// <param name="genericMsg">Generic message</param>
        /// <param name="cancelToken">Cancel token</param>
        /// <returns>Return TRUE if published successfully else FALSE</returns>
        public async Task<bool> PublishGenericAsync(TopicName topicName, GenericMessage genericMsg, CancellationToken cancelToken)
        {
            try
            {
                this.CheckProducerState();
                ValidateHelper.ValidateTopicName(topicName);
                ValidateHelper.ValidateGeneric(genericMsg);

                //// Produce
                cancelToken.ThrowIfCancellationRequested();
                var result = await this.ProduceGenericAsync(topicName, genericMsg, cancelToken).ConfigureAwait(false);

                return this.HandleDeliveryResult(result);
            }
            catch (OperationCanceledException ex)
            {
                Logger.Info(ex, ex.Message);
                return false;
            }
            catch (ProduceException<string, byte[]> ex)
            {
                Logger.Error(ex, $"{ex.Error.Reason}: {ex.Message}");
                return false;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }


        #region Commands

        /// <summary>
        /// Publish message
        /// </summary>
        /// <param name="topicName">Message topic</param>
        /// <param name="commandMsg">Command message</param>
        /// <param name="cancelToken">Cancel token</param>
        /// <returns>Return TRUE if published successfully else FALSE</returns>
        public bool PublishCommand(TopicName topicName, CommandMessage commandMsg, CancellationToken cancelToken)
        {
            try
            {
                this.CheckProducerState();
                ValidateHelper.ValidateTopicName(topicName);
                ValidateHelper.ValidateCommand(commandMsg);

                //// Produce
                cancelToken.ThrowIfCancellationRequested();
                return this.ProduceCommand(topicName, commandMsg, cancelToken);
            }
            catch (OperationCanceledException ex)
            {
                Logger.Info(ex, ex.Message);
                return false;
            }
            catch (ProduceException<string, byte[]> ex)
            {
                Logger.Error(ex, $"{ex.Error.Reason}: {ex.Message}");
                return false;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Publish command message asynchronously
        /// </summary>
        /// <param name="topicName">Message topic</param>
        /// <param name="commandMsg">Command message</param>
        /// <param name="cancelToken">Cancel token</param>
        /// <returns>Return TRUE if published successfully else FALSE</returns>
        public async Task<bool> PublishCommandAsync(TopicName topicName, CommandMessage commandMsg, CancellationToken cancelToken)
        {
            try
            {
                this.CheckProducerState();
                ValidateHelper.ValidateTopicName(topicName);
                ValidateHelper.ValidateCommand(commandMsg);

                //// Produce
                cancelToken.ThrowIfCancellationRequested();
                var result = await this.ProduceCommandAsync(topicName, commandMsg, cancelToken).ConfigureAwait(false);

                return this.HandleDeliveryResult(result);
            }
            catch (OperationCanceledException ex)
            {
                Logger.Info(ex, ex.Message);
                return false;
            }
            catch (ProduceException<string, byte[]> ex)
            {
                Logger.Error(ex, $"{ex.Error.Reason}: {ex.Message}");
                return false;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        #endregion Commands


        #region Events

        /// <summary>
        /// Publish event message
        /// </summary>
        /// <param name="topicName">Message topic</param>
        /// <param name="eventMsg">Event message</param>
        /// <param name="cancelToken">Cancel token</param>
        /// <returns>Return TRUE if published successfully else FALSE</returns>
        public bool PublishEvent(TopicName topicName, EventMessage eventMsg, CancellationToken cancelToken)
        {
            try
            {
                this.CheckProducerState();
                ValidateHelper.ValidateTopicName(topicName);
                ValidateHelper.ValidateEvent(eventMsg);

                //// Produce
                cancelToken.ThrowIfCancellationRequested();
                return this.ProduceEvent(topicName, eventMsg, cancelToken);
            }
            catch (OperationCanceledException ex)
            {
                Logger.Info(ex, ex.Message);
                return false;
            }
            catch (ProduceException<string, byte[]> ex)
            {
                Logger.Error(ex, $"{ex.Error.Reason}: {ex.Message}");
                return false;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Publish event message asynchronously
        /// </summary>
        /// <param name="topicName">Message topic</param>
        /// <param name="eventMsg">Event message</param>
        /// <param name="cancelToken">Cancel token</param>
        /// <returns>Return TRUE if published successfully else FALSE</returns>
        public async Task<bool> PublishEventAsync(TopicName topicName, EventMessage eventMsg, CancellationToken cancelToken)
        {
            try
            {
                this.CheckProducerState();
                ValidateHelper.ValidateTopicName(topicName);
                ValidateHelper.ValidateEvent(eventMsg);

                //// Produce
                cancelToken.ThrowIfCancellationRequested();
                var result = await this.ProduceEventAsync(topicName, eventMsg, cancelToken).ConfigureAwait(false);

                return this.HandleDeliveryResult(result);
            }
            catch (OperationCanceledException ex)
            {
                Logger.Info(ex, ex.Message);
                return false;
            }
            catch (ProduceException<string, byte[]> ex)
            {
                Logger.Error(ex, $"{ex.Message}");
                return false;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        #endregion Events


        /// <summary>
        /// Dispose
        /// </summary>
        public void Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        #endregion Interface Implementation


        #region Private Methods

        /// <summary>
        /// Builds the producer
        /// </summary>
        private void InitializeProducer()
        {
            var producerBuilder = new ProducerBuilder<string, byte[]>(this.producerConfig);
            producerBuilder.SetKeySerializer(Serializers.Utf8);
            producerBuilder.SetValueSerializer(Serializers.ByteArray);
            producerBuilder.SetErrorHandler(this.PublishErrorHandler);

            lock (this.mutex)
            {
                this.producer = producerBuilder.Build();
            }
        }


        #region Producer

        /// <summary>
        /// Produce command message
        /// </summary>
        /// <param name="topicName">Message topic</param>
        /// <param name="commandMsg">Command message</param>
        /// <param name="cancelToken">Cancel token</param>
        /// <returns>Return TRUE if published successfully else FALSE</returns>
        private bool ProduceGeneric(TopicName topicName, GenericMessage genericMsg, CancellationToken cancelToken)
        {
            return this.ProduceMessage<GenericMessage>(topicName, genericMsg.Data, genericMsg.GetAllHeadersAndProperties(), cancelToken);
        }

        /// <summary>
        /// Produce command message asynchronously
        /// </summary>
        /// <param name="topicName">Message topic</param>
        /// <param name="commandMsg">Command message</param>
        /// <param name="cancelToken">Cancel token</param>
        /// <returns>Return Kafka delivery result if published successfully</returns>
        private async Task<DeliveryResult<string, byte[]>> ProduceGenericAsync(TopicName topicName, GenericMessage genericMsg, CancellationToken cancelToken)
        {
            return await this.ProduceMessageAsync<GenericMessage>(topicName, genericMsg.Data, genericMsg.GetAllHeadersAndProperties(), cancelToken).ConfigureAwait(false);
        }

        /// <summary>
        /// Produce command message
        /// </summary>
        /// <param name="topicName">Message topic</param>
        /// <param name="commandMsg">Command message</param>
        /// <param name="cancelToken">Cancel token</param>
        /// <returns>Return TRUE if published successfully else FALSE</returns>
        private bool ProduceCommand(TopicName topicName, CommandMessage commandMsg, CancellationToken cancelToken)
        {
            return this.ProduceMessage<CommandMessage>(topicName, commandMsg.Data, commandMsg.GetAllHeadersAndProperties(), cancelToken);
        }

        /// <summary>
        /// Produce command message asynchronously
        /// </summary>
        /// <param name="topicName">Message topic</param>
        /// <param name="commandMsg">Command message</param>
        /// <param name="cancelToken">Cancel token</param>
        /// <returns>Return Kafka delivery result if published successfully</returns>
        private async Task<DeliveryResult<string, byte[]>> ProduceCommandAsync(TopicName topicName, CommandMessage commandMsg, CancellationToken cancelToken)
        {
            return await this.ProduceMessageAsync<CommandMessage>(topicName, commandMsg.Data, commandMsg.GetAllHeadersAndProperties(), cancelToken).ConfigureAwait(false);
        }

        /// <summary>
        /// Produce event message
        /// </summary>
        /// <param name="topicName">Message topic</param>
        /// <param name="eventMsg">Event message</param>
        /// <param name="cancelToken">Cancel token</param>
        /// <returns>Return TRUE if published successfully else FALSE</returns>
        private bool ProduceEvent(TopicName topicName, EventMessage eventMsg, CancellationToken cancelToken)
        {
            return this.ProduceMessage<EventMessage>(topicName, eventMsg.Data, eventMsg.GetAllHeadersAndProperties(), cancelToken);
        }

        /// <summary>
        /// Produce event message asynchronously
        /// </summary>
        /// <param name="topicName">Message topic</param>
        /// <param name="eventMsg">Event message</param>
        /// <param name="cancelToken">Cancel token</param>
        /// <returns>Return Kafka delivery result if published successfully</returns>
        private async Task<DeliveryResult<string, byte[]>> ProduceEventAsync(TopicName topicName, EventMessage eventMsg, CancellationToken cancelToken)
        {
            return await this.ProduceMessageAsync<EventMessage>(topicName, eventMsg.Data, eventMsg.GetAllHeadersAndProperties(), cancelToken).ConfigureAwait(false);
        }

        /// <summary>
        /// Produce message
        /// </summary>
        /// <typeparam name="TMessage">Message data type</typeparam>
        /// <param name="topicName">Message topic</param>
        /// <param name="data">Message data</param>
        /// <param name="headerMaps">Meta-data for message</param>
        /// <param name="cancelToken">Cancel token</param>
        /// <returns>Return TRUE if published successfully else FALSE</returns>
        private bool ProduceMessage<TMessage>(TopicName topicName, byte[] data, IDictionary<string, string> headerMaps, CancellationToken cancelToken)
        {
            var kafkaMessage = CreateKafkaMessage<TMessage>(topicName, data, headerMaps);
            this.producer.Produce(topicName.Name.Trim(), kafkaMessage, this.HandleDeliveryReport);

            //// Flush the synchronous producer send call
            if (cancelToken.CanBeCanceled)
            {
                this.producer.Flush(TimeoutProducerFlush);
            }
            else
            {
                this.producer.Flush(cancelToken);
            }

            return true;
        }

        /// <summary>
        /// Produce message asynchronously
        /// </summary>
        /// <typeparam name="TMessage">Message data type</typeparam>
        /// <param name="topicName">Message topic</param>
        /// <param name="data">Message data</param>
        /// <param name="headerMaps">Meta-data for message</param>
        /// <param name="cancelToken">Cancel token</param>
        /// <returns>Return Kafka delivery result if published successfully</returns>
        private async Task<DeliveryResult<string, byte[]>> ProduceMessageAsync<TMessage>(TopicName topicName, byte[] data, IDictionary<string, string> headerMaps, CancellationToken cancelToken)
        {
            var kafkaMessage = CreateKafkaMessage<TMessage>(topicName, data, headerMaps);

            return await this.producer.ProduceAsync(topicName.Name.Trim(), kafkaMessage, cancelToken).ConfigureAwait(false);
        }

        #endregion Producer


        /// <summary>
        /// Handle delivery report callback.
        /// Callback for synchronous publish.
        /// </summary>
        /// <param name="report">Delivery report</param>
        private void HandleDeliveryReport(DeliveryReport<string, byte[]> report)
        {
            if (report != null)
            {
                var outcome = CreatePublishOutcome(report.Message.Value, report);
                this.publishResponseHandler?.Invoke(outcome);

                if (report.Error == null || report.Error.Code == ErrorCode.NoError)
                {
                    Logger.Info("Message published successfully!");
                }
                else
                {
                    Logger.Error($"Error encountered on publishing message: {report.Error.Code}");
                }
            }
        }

        /// <summary>
        /// Handle delivery result callback.
        /// Callback for asynchronous publish.
        /// </summary>
        /// <param name="result">Publish delivery result</param>
        /// <returns>Return TRUE if delivery response handled successfully else FALSE</returns>
        private bool HandleDeliveryResult(DeliveryResult<string, byte[]> result)
        {
            bool? isHandled = false;

            if (result != null)
            {
                if (result.Status == PersistenceStatus.NotPersisted)
                {
                    Logger.Warn($"[Topic: {result.Topic}, {result.Status}] Error encountered on publishing message.");
                }
                else
                {
                    var outcome = CreatePublishOutcome(result.Message.Value, result, null);
                    isHandled = this.publishResponseHandler?.Invoke(outcome);
                }
            }

            return isHandled ?? false;
        }

        /// <summary>
        /// Set the handler to call on error events e.g. connection failures or all brokers down.
        /// Note that the client will try to automatically recover from errors that are not marked as fatal.
        /// Non-fatal errors should be interpreted as informational rather than catastrophic.
        /// </summary>
        /// <param name="iproducer">Producer instance</param>
        /// <param name="error">Kafka error</param>
        private void PublishErrorHandler(IProducer<string, byte[]> iproducer, Error error)
        {
            this.publishErrorHandler?.Invoke(error.Code.ToString(), error.Reason, error.IsFatal);
        }

        /// <summary>
        /// Check current producer state
        /// </summary>
        private void CheckProducerState()
        {
            if (!this.IsInitialized)
            {
                throw new InvalidOperationException($"Producer is not initialized. IsInit: {this.IsInitialized}");
            }
            else if (this.IsDisposed || this.producer == null)
            {
                Logger.Error("Publisher have been disposed, create new instance required.");
                throw new TypeInitializationException(typeof(PublisherImpl).FullName, null);
            }
        }

        /// <summary>
        /// Dispose producer
        /// </summary>
        private void DisposeProducer()
        {
            lock (this.mutex)
            {
                if (this.producer != null)
                {
                    this.producer.Flush();
                    this.producer.Dispose();
                    this.producer = null;
                }
            }
        }

        /// <summary>
        /// Dispose
        /// </summary>
        /// <param name="disposing">If disposing equals true, the method has been called directly
        /// or indirectly by a user's code. Managed and unmanaged resources can be disposed.
        /// If disposing equals false, the method has been called by the runtime from inside the
        /// finalizer and you should not reference other objects. Only unmanaged resources can be disposed.</param>
        private void Dispose(bool disposing)
        {
            try
            {
                if (!this.IsDisposed && disposing)
                {
                    this.DisposeProducer();
                }

                this.IsDisposed = true;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        #endregion Private Methods


        #region Static Methods

        /// <summary>
        /// Create kafka message for publish
        /// </summary>
        /// <typeparam name="TMessage">Message data type</typeparam>
        /// <param name="topicName">Message topic</param>
        /// <param name="messageData">Message data</param>
        /// <param name="headerMaps">Meta-data for message</param>
        /// <returns>Return new kafka message</returns>
        private static Message<string, byte[]> CreateKafkaMessage<TMessage>(TopicName topicName, byte[] messageData, IDictionary<string, string> headerMaps)
        {
            var kafkaMessage = new Message<string, byte[]>
            {
                Key = topicName.Route?.Trim(),
                Value = messageData,
                Headers = GenerateHeaders(headerMaps, typeof(TMessage).FullName),
                Timestamp = new Timestamp(DateTime.UtcNow),
                ////Timestamp = Timestamp.Default, // Note: Do NOT use this timestamp, DT type is unspecified (neither UTC nor local).
            };

            return kafkaMessage;
        }

        /// <summary>
        /// Generate Kafka message header
        /// </summary>
        /// <param name="properties">Meta-data for message</param>
        /// <param name="messageType">Message/payload type</param>
        /// <returns>Return new Kafka message header</returns>
        private static Headers GenerateHeaders(IDictionary<string, string> properties, string messageType)
        {
            var headers = new Headers();

            //// Create headers for properties
            if (properties?.Any() == true)
            {
                foreach (var item in properties)
                {
                    //// Serialize and create new header
                    var byteValue = JsonSerializer.Serialize(item.Value);
                    headers.Add(new Header(item.Key, byteValue));
                }
            }

            //// Create header for main message type
            headers.Add(new Header(HeaderKeys.HeaderKeyMessageType, SerializeString(messageType, Encoding.UTF8)));

            return headers;
        }

        /// <summary>
        /// Create message publish outcome
        /// </summary>
        /// <param name="msg">Published message</param>
        /// <param name="report">Delivery report</param>
        /// <returns>Return message publish outcome</returns>
        private static PublishOutcome CreatePublishOutcome(byte[] msg, DeliveryReport<string, byte[]> report)
        {
            return CreatePublishOutcome(msg, report, report.Error);
        }

        /// <summary>
        /// Create message publish outcome
        /// </summary>
        /// <param name="msg">Published message</param>
        /// <param name="result">Delivery result</param>
        /// <param name="error">Publish error if any</param>
        /// <returns>Return message publish outcome</returns>
        private static PublishOutcome CreatePublishOutcome(byte[] msg, DeliveryResult<string, byte[]> result, Error error)
        {
            var persistState = (result.Status == PersistenceStatus.Persisted) ? PersistenceState.Persisted : PersistenceState.NotPersisted;
            var publishedDT = result.Timestamp.UtcDateTime.ToLocalTime();
            var hasError = !(error == null || error?.Code == ErrorCode.NoError);
            var errorReason = error?.Reason;

            return new PublishOutcome(msg, persistState, publishedDT, hasError, errorReason);
        }

        /// <summary>
        /// Serialize string
        /// </summary>
        /// <param name="str">String to serialize</param>
        /// <returns>Return serialized string in byte array</returns>
        private static byte[] SerializeString(string str, Encoding encoding)
        {
            var tempStr = str?.Trim();

            //// Continue serialization even if string is empty
            return tempStr == null ? null : encoding.GetBytes(str);
        }

        #endregion Static Methods
    }
}
